package com.bw.gmall.realtime.app.ods;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.File;
import java.io.FileNotFoundException;
import java.util.*;

public class BaseLogAppShopOds {
    public static void main(String[] args) throws InterruptedException, FileNotFoundException {
        Long [] arrTs={
                1730307600000l, 1730311200000l, 1730314800000l, 1730318400000l, 1730321400000l, 1730324400000l, 1730328400000l, 1730331400000l, 1730334400000l, 1730338400000l, 1730341400000l, 1730344400000l, 1730348400000l, 1730351400000l,
                1730354400000l, 1730358400000l, 1730361400000l, 1730364400000l, 1730368400000l, 1730371800000l, 1730374100000l, 1730378400000l, 1730381800000l, 1730384800000l, 1730388800000l
        };
        ArrayList<String> list = new ArrayList<>();
        list.add("o");
        list.add("a");
        list.add("b");
        list.add("c");
        list.add("d");
        list.add("e");

        for (Long arrT:arrTs){
            mockDatas(list,arrT);
            System.out.println("-------------->");
            Thread.sleep(3000);
        }
        }
    public static void mockDatas(ArrayList<String> list,Long ts) throws FileNotFoundException {
        ArrayList<JSONObject> jsonObjects = new ArrayList<>();
        File file = new File("datas/app.2025-06-23.log");
        Scanner scanner = new Scanner(file);
        while(scanner.hasNextLine()){
            String line = scanner.nextLine();
            JSONObject jsonObj = JSON.parseObject(line);
            JSONObject common = jsonObj.getJSONObject("common");
            JSONObject page = jsonObj.getJSONObject("page");
            common.put("shop_id",new Random().nextInt(10)+1);
            common.put("mid",new Random().nextInt(100000)+1);
            int i = new Random().nextInt(2);
            int i1 = new Random().nextInt(5);
            if(page!=null){
                page.put("page_action",list.get(i1));
            }
            common.put("device_type",(i==0?0:1));
            jsonObj.put("ts",ts);
            jsonObjects.add(jsonObj);
        }
     Map<String, Object> config = new HashMap<>();
        config.put("bootstrap.servers","hadoop102:9092");
        config.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        config.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");

        //创建kafka生产者
        Producer<String, String> producer = new KafkaProducer<>(config);

        //创建JSON对象
        ObjectMapper mapper = new ObjectMapper();

        try {
            //假设我们有一个JSON字符串
            for (JSONObject jsonObject : jsonObjects) {
                //发送到kafka
                producer.send(new ProducerRecord<>("ods_traffic",jsonObject.toJSONString()));
            }
        } catch (Exception e) {
           e.printStackTrace();
        }finally {
            producer.close();
        }
        scanner.close();
    }
}
